package com.swak.paxos.impl;

import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import com.swak.paxos.Learner;
import com.swak.paxos.PaxosConfig;
import com.swak.paxos.Value;
import com.swak.paxos.breakpoint.LearnerBP;
import com.swak.paxos.enums.ProposalType;
import com.swak.paxos.protocol.PaxosMessage;
import com.swak.paxos.state.BallotNumber;
import com.swak.paxos.state.LearnerState;
import com.swak.paxos.store.PaxosLog;
import com.swak.paxos.store.PaxosStorage;
import com.swak.reactivex.threads.EventLoop;

import lombok.Getter;

/**
 * PaxosLearner.
 * 
 * 数据对齐
 * 
 * @author 618lf
 */
@Getter
public class PaxosLearner extends AutomaticLearner implements Learner {

	private final PaxosNode node;
	private final PaxosGroup group;
	private final PaxosConfig config;
	private final PaxosInstance instance;
	private final PaxosAcceptor acceptor;
	private final EventLoop eventLoop;
	private final PaxosStorage storage;

	/**
	 * 记录 Preparing 或 Accepting 的超时任务.
	 */
	private ScheduledFuture<?> timeoutFuture;
	private final Runnable askforLearnTimeout;

	/**
	 * 当前正在进行的步骤
	 */
	private final LearnerState state = new LearnerState();
	private boolean isIMLearning;

	/**
	 * 当前收到的最大的值ID
	 */
	private long highestSeenInstanceID;
	private String highestSeenInstanceIDFromNodeID;

	public PaxosLearner(PaxosConfig config, PaxosNode node, PaxosGroup group, PaxosInstance instance,
			PaxosAcceptor acceptor, EventLoop eventLoop, PaxosStorage storage) {
		super(config);
		this.config = config;
		this.node = node;
		this.group = group;
		this.instance = instance;
		this.acceptor = acceptor;
		this.eventLoop = eventLoop;
		this.storage = storage;
		this.askforLearnTimeout = this::askforLearnNoop;
	}

	@Override
	public void newInstance() {
		this.state.init();
	}

	@Override
	public boolean isLearned() {
		return this.state.isLearned();
	}

	@Override
	public Value getLearnedValue() {
		return this.state.getLearnedValue();
	}

	// 第一次收到最新的数据
	@Override
	public void onProposed(PaxosMessage message) {

		LearnerBP.onProposed(this.group.getGroupID(), this.instance.getInstanceID());

		/*
		 * 需要校验 Acceptor 当前的状态，是否真实的接收了最新的值
		 */
		if (!this.acceptor.checkAcceptBallot(message)) {
			LearnerBP.onProposedBallotNotSame(this.group.getGroupID(), this.instance.getInstanceID());
			return;
		}

		/*
		 * 设置 Learner 状态
		 */
		this.state.learnValueWithoutWrite(this.acceptor.getState().getAcceptedValue());

		LearnerBP.onProposedSuccessLearn(this.group.getGroupID(), this.instance.getInstanceID());

		/*
		 * 发送给 Follower
		 */
		if (this.group.isIMMaster()) {
			this.transmitToFollower();
		}
	}

	private void transmitToFollower() {

		/*
		 * 准备消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setType(ProposalType.SendLearnValue);
		message.setGroupId(this.group.getGroupID());
		message.setInstanceID(this.instance.getInstanceID());
		message.setNodeID(this.node.id());
		message.setProposalNodeID(this.acceptor.getState().getAcceptedBallot().getNodeID());
		message.setProposalID(this.acceptor.getState().getAcceptedBallot().getProposalID());
		message.setValue(this.acceptor.getState().getAcceptedValue());

		/*
		 * 通过 TCP 的方式发送
		 */
		this.node.broadcastMessageToFollower(message);
	}

	// 第二次收到最新的数据
	@Override
	public void onSendLearnValue(PaxosMessage message) {

		LearnerBP.onSendLearnValue(this.group.getGroupID());

		/*
		 * 不能学习这个值
		 */
		if (message.getInstanceID() > this.instance.getInstanceID()) {
			return;
		}

		/*
		 * 否则不需要学习这个值， 但是如果要回复消息则也需要回复
		 */
		if (message.getInstanceID() == this.instance.getInstanceID()) {

			/*
			 * 存储
			 */
			BallotNumber ballot = new BallotNumber(message.getProposalID(), message.getProposalNodeID());
			boolean success = this.persist(message.getInstanceID(), ballot, message.getValue());
			if (!success) {
				return;
			}

			this.state.learnValueWithoutWrite(message.getValue());
		}

		/*
		 * 是否需要回复学习结果
		 */
		if (message.getType() == ProposalType.SendLearnValueNeedAck) {
			this.resetAskforLearnTimeout();
			this.sendLearnValueAck(message.getNodeID());
		}
	}

	private void resetAskforLearnTimeout() {
		this.cancelTimeout();
		this.timeoutFuture = this.eventLoop.schedule(askforLearnTimeout, this.config.getAskForLearnTimeoutMs(),
				TimeUnit.MILLISECONDS);
	}

	void cancelTimeout() {
		if (this.timeoutFuture != null && !this.timeoutFuture.isCancelled() && !this.timeoutFuture.isDone()) {
			this.timeoutFuture.cancel(false);
		}
		this.timeoutFuture = null;
	}

	private void sendLearnValueAck(String askNodeID) {

		LearnerBP.sendLearnValueAck(this.group.getGroupID());

		/*
		 * 回复消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setInstanceID(this.instance.getInstanceID());
		message.setType(ProposalType.SendLearnValueAck);
		message.setNodeID(this.node.id());
		this.node.sendMessage(askNodeID, message);
	}

	private boolean persist(long instanceID, BallotNumber ballot, Value value) {
		PaxosLog log = new PaxosLog();
		log.setInstanceID(instanceID);
		log.setAcceptedValue(value);
		log.setPromiseID(ballot.getProposalID());
		log.setPromiseNodeID(ballot.getNodeID());
		log.setAcceptedID(ballot.getProposalID());
		log.setAcceptedNodeID(ballot.getNodeID());
		return this.storage.storage(log);
	}

	// 第二次收到最新的数据 -- 处理确认
	@Override
	public void onSendLearnValueAck(PaxosMessage message) {
		LearnerBP.onSendLearnValueAck(this.group.getGroupID());

		// 这里处理 Ack

	}

	// 循环：Learner 发起学习
	void askforLearnNoop() {

		/*
		 * 重新添加此超时任务.
		 */
		this.resetAskforLearnTimeout();

		/*
		 * 开启学习
		 */
		this.askforLearn();
	}

	private void askforLearn() {

		LearnerBP.askforLearn(this.group.getGroupID());

		/*
		 * 准备消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setType(ProposalType.AskforLearn);
		message.setGroupId(this.group.getGroupID());
		message.setInstanceID(this.instance.getInstanceID());
		message.setNodeID(this.node.id());

		/*
		 * 发送消息
		 */
		this.node.broadcastMessage(message);
	}

	@Override
	public void onAskforLearn(PaxosMessage message) {

		LearnerBP.onAskforLearn(this.group.getGroupID());

		this.setSeenInstanceID(message.getInstanceID(), message.getNodeID());

		// 不需要发送学习数据
		if (message.getInstanceID() >= this.instance.getInstanceID()) {
			return;
		}

		// 如果 需要的值
		this.sendNowInstanceID(message.getInstanceID(), message.getNodeID());
	}

	private void setSeenInstanceID(long instanceID, String fromNodeID) {
		if (instanceID > this.highestSeenInstanceID) {
			this.highestSeenInstanceID = instanceID;
			this.highestSeenInstanceIDFromNodeID = fromNodeID;
		}
	}

	private void sendNowInstanceID(long instanceID, String nodeID) {

		LearnerBP.sendNowInstanceID(this.group.getGroupID());

		/*
		 * 准备消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setType(ProposalType.SendNowInstanceID);
		message.setGroupId(this.group.getGroupID());
		message.setInstanceID(instanceID);
		message.setNodeID(this.node.id());
		message.setNowInstanceID(this.instance.getInstanceID());

		// 发送消息
		this.node.sendMessage(nodeID, message);
	}

	@Override
	public void onSendNowInstanceID(PaxosMessage message) {
		LearnerBP.onSendNowInstanceID(this.group.getGroupID());
		long receiveNowInstanceId = message.getNowInstanceID();
		long currInstanceId = this.instance.getInstanceID();
		setSeenInstanceID(receiveNowInstanceId, message.getNodeID());

		// 掉队的消息，可以忽略
		if (currInstanceId != message.getInstanceID()) {
			return;
		}

		// 掉队的消息，可以忽略
		if (receiveNowInstanceId <= currInstanceId) {
			return;
		}

		// 正在学习中，可以忽略
		if (this.isIMLearning) {
			return;
		}

		// 确认从 NodeID 中学习
		comfirmAskForLearn(message.getNodeID());
	}

	private void comfirmAskForLearn(String nodeID) {

		LearnerBP.comfirmAskForLearn(this.group.getGroupID());

		/*
		 * 准备消息
		 */
		PaxosMessage message = new PaxosMessage();
		message.setType(ProposalType.ComfirmAskforLearn);
		message.setGroupId(this.group.getGroupID());
		message.setNodeID(this.node.id());
		message.setInstanceID(this.instance.getInstanceID());

		/*
		 * 发送消息
		 */
		this.node.sendMessage(nodeID, message);

		/*
		 * 消息发送成功，进入数据对齐阶段
		 */
		this.isIMLearning = true;
	}

	@Override
	public void onComfirmAskforLearn(PaxosMessage message) {

		LearnerBP.onComfirmAskForLearn(this.group.getGroupID());

	}
}
